package cn.itcast.flink.connector;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

import static org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS;

/**
 * Author itcast
 * Date 2022/1/13 15:03
 * Desc 消费或读取Kafka中的数据
 * 开发步骤：
 * 1.获取流执行环境
 * 2.设置并行度
 * 3.Flink消费Kafka数据并设置相关参数
 * 3.1 配置参数属性
 *      主机列表，topic,消费者组,自动发现分区参数
 * 3.2 实例化 FlinkKafkaConsumer 对象来进行消费数据
 * 4.设置消费Kafka的重启位置 earliest latest specified offset timestamp groupOffset
 * 5.将 offset 提交给checkpoint维护和管理
 * 6.将 consumer 添加到 flink source 中
 * 7. 打印输出
 * 8.执行流环境
 */
public class FlinkKafkaReader {
    public static void main(String[] args) throws Exception {
        //1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        //大家去公司，版本是 Flink1.12- 需要设置TimeCharacteristic
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(1000);
        //3.Flink消费Kafka数据并设置相关参数
        //3.1 配置参数属性
        //     主机列表，topic,消费者组,自动发现分区参数
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"_consumer_kafka_");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
        //自动分区发现参数：能够定时捕获到Kafka Partition 3=>6
        //The default interval to execute partition discovery, in milliseconds ({@code Long.MIN_VALUE},
        //i.e. disabled by default).
        props.setProperty(KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,"6000");
        //3.2 实例化 FlinkKafkaConsumer 对象来进行消费数据
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "writetopic",
                new SimpleStringSchema(),
                props
        );
        //4.设置消费Kafka的重启位置
        // earliest: 从最早的 offset
        // latest： 从最新的
        // specified offset：从指定的offset
        // timestamp：从指定的时间恢复
        // groupOffset： 从消费组中内的 offset 进行消费（默认）
        consumer.setStartFromLatest();
        //5.将 offset 提交给checkpoint维护和管理
        //如果将 offset 托管给 flink checkpoint 去管理，需要先开启 checkpoint 才能提交和恢复；
        //如果没有开启 checkpoint ，就会根据 enable.auto.commit 到 kafka 消费者组中
        consumer.setCommitOffsetsOnCheckpoints(true);
        //6.将 consumer 添加到 flink source 中
        DataStreamSource<String> source = env.addSource(consumer);
        //7. 打印输出
        source.printToErr();
        //8.执行流环境
        env.execute();
    }
}
